package com.dtwave.lnstreaming.process;

import com.alibaba.fastjson.JSONObject;
import com.dtwave.lnstreaming.bean.RtDkhtxxb;
import com.dtwave.lnstreaming.connection.ElasticConnector;
import com.dtwave.lnstreaming.filter.XfaceFilterFunction;
import com.dtwave.lnstreaming.filter.function.FlatMapFunction;
import com.dtwave.lnstreaming.filter.function.NotNullFilterFunction;
import com.dtwave.lnstreaming.process.function.process.XfaceFlatFunction;
import com.dtwave.lnstreaming.process.function.process.XfaceKeyProcessFunction;
import com.dtwave.lnstreaming.process.selector.XfaceKeySelector;
import com.dtwave.lnstreaming.sink.EsSinkFunctionContractInfo;
import com.dtwave.param.EsparamObj;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.connectors.elasticsearch7.ElasticsearchSink;
import org.apache.flink.table.planner.expressions.In;
import org.apache.hadoop.hbase.ClassFinder;

public class StreamDataExecutor {

    public static void executeFOResult(SingleOutputStreamOperator<JSONObject> flatMapStream, EsparamObj esParamObj) {
        //财富总览-贷款明细表

        //财富总览-贷款详情表

        //贷款管理-贷款合同表

        //贷款管理-贷款合同信息表
        ElasticsearchSink.Builder<RtDkhtxxb> esBuilderContractInfo = getEsBuilderContractInfo(esParamObj);

        SingleOutputStreamOperator<RtDkhtxxb> xfaceExpandStream = flatMapStream
                .filter(new XfaceFilterFunction())
                .keyBy(new XfaceKeySelector())
                .process(new XfaceKeyProcessFunction())
                .flatMap(new XfaceFlatFunction());

        xfaceExpandStream
                .filter(new NotNullFilterFunction<>())
                .addSink(esBuilderContractInfo.build())
                .name("DKHTXXB SINK");

    }

    private static ElasticsearchSink.Builder<RtDkhtxxb> getEsBuilderContractInfo(EsparamObj esParamObj) {
        ElasticsearchSink.Builder<RtDkhtxxb> esBuilder = new ElasticsearchSink.Builder<>(
                ElasticConnector.parse(esParamObj.getEsUrl()),
                new EsSinkFunctionContractInfo(esParamObj)
        );
        esBuilder.setBulkFlushInterval(Integer.parseInt(esParamObj.getEsIntervalMs()));
        esBuilder.setBulkFlushMaxActions(Integer.parseInt(esParamObj.getEsMaxAction()));
        esBuilder.setBulkFlushBackoffRetries(Integer.parseInt(esParamObj.getEsMaxReties()));
        return esBuilder;


    }
}
